//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// hash_join_executor.h
//
// Identification: src/include/execution/executors/hash_join_executor.h
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#pragma once

#include <memory>
#include <utility>

#include "common/util/hash_util.h"
#include "execution/executor_context.h"
#include "execution/executors/abstract_executor.h"
#include "execution/plans/hash_join_plan.h"
#include "storage/table/tuple.h"

namespace bustub {

struct JoinKey {
  std::vector<Value> keys_;
  auto operator==(const JoinKey &other) const -> bool {
    for (uint32_t i = 0; i < other.keys_.size(); i++) {
      if (keys_[i].CompareEquals(other.keys_[i]) != CmpBool::CmpTrue) {
        return false;
      }
    }
    return true;
  }
};

struct JoinValue {
  std::vector<std::pair<JoinKey, Tuple>> vals;
};

}  // namespace bustub

namespace std {

template <>
struct hash<bustub::JoinKey> {
  auto operator()(const bustub::JoinKey &agg_key) const -> std::size_t {
    size_t curr_hash = 0;
    for (const auto &key : agg_key.keys_) {
      if (!key.IsNull()) {
        curr_hash = bustub::HashUtil::CombineHashes(curr_hash, bustub::HashUtil::HashValue(&key));
      }
    }
    return curr_hash;
  }
};
}  // namespace std

namespace bustub {
/** Implements std::hash on AggregateKey */
class JoinHashTable {
 public:
  void Insert(const Tuple &tuple, const Schema &schema, const std::vector<AbstractExpressionRef> &exprs) {
    JoinKey key = GetHashKey(tuple, schema, exprs);
    ht_[key].vals.emplace_back(key, tuple);
  }
  auto GetValue(const Tuple &tuple, const Schema &schema, const std::vector<AbstractExpressionRef> &exprs,
                JoinValue &join_value) -> bool {
    JoinKey key = GetHashKey(tuple, schema, exprs);
    if (ht_.find(key) == ht_.end()) {
      return false;
    }
    join_value = ht_[key];
    return true;
  }
  static auto GetHashKey(const Tuple &tuple, const Schema &schema, const std::vector<AbstractExpressionRef> &exprs)
      -> JoinKey {
    std::vector<Value> keys;
    keys.reserve(exprs.size());
    for (const auto &key_expr : exprs) {
      keys.push_back(key_expr->Evaluate(&tuple, schema));
    }
    return JoinKey{keys};
  }

  void Clear() { ht_.clear(); }

 private:
  /** The hash table is just a map from aggregate keys to aggregate values */
  std::unordered_map<JoinKey, JoinValue> ht_{};
};
/**
 * HashJoinExecutor executes a nested-loop JOIN on two tables.
 */
class HashJoinExecutor : public AbstractExecutor {
 public:
  /**
   * Construct a new HashJoinExecutor instance.
   * @param exec_ctx The executor context
   * @param plan The HashJoin join plan to be executed
   * @param left_child The child executor that produces tuples for the left side of join
   * @param right_child The child executor that produces tuples for the right side of join
   */
  HashJoinExecutor(ExecutorContext *exec_ctx, const HashJoinPlanNode *plan,
                   std::unique_ptr<AbstractExecutor> &&left_child, std::unique_ptr<AbstractExecutor> &&right_child);

  /** Initialize the join */
  void Init() override;

  /**
   * Yield the next tuple from the join.
   * @param[out] tuple The next tuple produced by the join.
   * @param[out] rid The next tuple RID, not used by hash join.
   * @return `true` if a tuple was produced, `false` if there are no more tuples.
   */
  auto Next(Tuple *tuple, RID *rid) -> bool override;

  /** @return The output schema for the join */
  auto GetOutputSchema() const -> const Schema & override { return plan_->OutputSchema(); };

 private:
  /** The NestedLoopJoin plan node to be executed. */
  const HashJoinPlanNode *plan_;
  std::unique_ptr<AbstractExecutor> lch_;
  std::unique_ptr<AbstractExecutor> rch_;
  JoinHashTable ht_{};
  Tuple left_tuple_{};
  JoinValue cur_join_value_{};
  size_t cur_idx_{0};
  bool has_match_{false};

  auto JoinInner(Tuple *tuple) -> bool;
  auto JoinLeft(Tuple *tuple) -> bool;
  void GenerateNullTuple(Tuple *);
  void GenerateGoodTuple(Tuple *, const Tuple &);
};

}  // namespace bustub
